package org.example.listener;

import com.alibaba.fastjson2.JSONObject;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import com.alibaba.otter.canal.protocol.FlatMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @author shenguangyang
 */
@Component
public class CanalListener {
    private static final Logger log = LoggerFactory.getLogger(CanalListener.class);

    @KafkaListener(
            groupId = "house-consumer-group",
            topicPartitions = {@TopicPartition(topic = "house-topic", partitions = {"0"})}
    )
    public void partition1(ConsumerRecord<String, String> record, Acknowledgment ack) {
        receive(record, ack);
    }

    @KafkaListener(
            groupId = "house-consumer-group",
            topicPartitions = {@TopicPartition(topic = "house-topic", partitions = {"1"})}
    )
    public void partition2(ConsumerRecord<String, String> record, Acknowledgment ack) {
        receive(record, ack);
    }

    @KafkaListener(
            groupId = "house-consumer-group",
            topicPartitions = {@TopicPartition(topic = "house-topic", partitions = {"2"})}
    )
    public void partition3(ConsumerRecord<String, String> record, Acknowledgment ack) {
        receive(record, ack);
    }

    private void receive(ConsumerRecord<String, String> record, Acknowledgment ack) {
        final String value = record.value();
        FlatMessage message = JSONObject.parseObject(value, FlatMessage.class);
        final String houseCode = message.getData().get(0).get("house_code");
        System.out.println("分区:" + record.partition() + "\t接收到数据的code: " + houseCode + "\t操作类别: " + message.getType());
        ack.acknowledge();
    }
}
